package demo.hello;

import java.util.Date;
import java.util.List;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

/*
 * 推模式的消费者,应用被动接收服务端通知,
 * 使用listener的方式监听消息
 * 这是个比较好的方式
 */
public class ListenerConsumerUser {
    public static void main(String[] args) throws MQClientException {
        // 1 创建一个拉取消息的消费者对象
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("demo1");
        // 2 设置名字地址
        consumer.setNamesrvAddr("10.1.5.17:9876");
        // 3 绑定消息的主题
        consumer.subscribe("TopicTest1", "*");
        // 4 消费者监听处理消息的方法
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list,
                    ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (MessageExt msg : list) {
                    System.out.println(new Date()+"消费线程:" + Thread.currentThread().getName() + ",消息ID:" + msg.getMsgId());
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 启动消费者
        consumer.start();
    }
}
